package zy.learn.demo.structuredstreaming.sink

import org.apache.spark.SparkConf
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object KafkaSink {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().set("spark.sql.shuffle.partitions", "3")
    val spark: SparkSession = SparkSession
      .builder()
      .config(sparkConf)
      .master("local[1]")
      .appName("Test")
      .getOrCreate()
    import spark.implicits._

    val lines: DataFrame = spark.readStream
      .format("socket") // 设置数据源
      .option("host", "co7-203")
      .option("port", 10000)
      .load

    val words = lines.as[String]
      .flatMap(_.split("\\W+"))
      .groupBy("value")
      .count()
      .map(row => row.getString(0) + "," + row.getLong(1))
      .toDF("value")  // 写入数据时候, 必须有一列 "value"

    words.writeStream
      .outputMode("update")
      .format("kafka")
      .trigger(Trigger.ProcessingTime(0))
      .option("kafka.bootstrap.servers", "co7-203:9092,co7-204:9092,co7-205:9092") // kafka 配置
      .option("topic", "update") // kafka 主题
      .option("checkpointLocation", "./ck1")  // 必须指定 checkpoint 目录
      .start
      .awaitTermination()
  }
}
